package com.gugee.ins.data.task.analysis.service;

import com.gugee.common.task.dto.TaskParam;
import com.gugee.common.task.model.TaskRecord;
import com.gugee.common.task.service.AbstractProducerHandler;
import com.gugee.ins.common.constant.RedisKeyConst;
import com.gugee.ins.common.dal.service.blogger.InsBloggerMainDao;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import com.gugee.ins.data.task.config.TaskConfig;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.List;

/**
 * 分析账户是否异常信息发送
 * @author Huangxin
 */
@Component
@Slf4j
public class InsAnalysisAccountCheckUnusualSvc extends AbstractProducerHandler<InsBloggerMessage> {
    /**
     *
     */
    @Autowired
    @Qualifier("insBloggerMsgBaseInfoProducer")
    private KafkaTemplate<Long, InsBloggerMessage> insBloggerMsgBaseInfoProducer;

    /**
     *
     */
    @Autowired
    public InsBloggerMainDao insBloggerMainDao;

    /**
     * redis
     */
    @Autowired
    RedisTemplate<String,String> redisTemplate;

    /**
     *
     */
    @Autowired
    public TaskConfig taskConfig;

    /**
     * 分析账户是否异常信息发送
     * @param taskRecord 业务类型数据的采集页
     * @param taskParam  任务参数，包含了每页的页码大小
     * @return
     */
    @Override
    public List<InsBloggerMessage> findMessageObject(TaskRecord taskRecord, TaskParam taskParam) {
        // 处理的条数
        int count = taskParam.getRecords();
        // 需要采集的数据
        List<InsBloggerMessage> rst = new ArrayList<>(count);
        String value = "";
        // 从redis中获取前几条数据
        for(int i = 0; i < count; ++i){
            try{
                value = redisTemplate.opsForList().leftPop(RedisKeyConst.INS_SPIDER_UNUSUAL_LIST);
                if(!StringUtils.isEmpty(value)){
                    InsBloggerMessage insBloggerMessage = new InsBloggerMessage();
                    insBloggerMessage.setUid(Long.parseLong(value));
                    insBloggerMessage.setRetryTimes(taskConfig.getInsAnalysisAccountCheckUnusualRetryTimes());
                    insBloggerMessage.setRetryLimitTimes(taskConfig.getInsAnalysisAccountCheckUnusualRetryTimes());
                    rst.add(insBloggerMessage);
                }
            }catch (Exception e){
                log.error("从redis获取异常的uid error,uid:{},msg:{}",value,e.getMessage());
            }
        }
        return rst;
    }

    /**
     * 发送
     * @param taskType 任务类型
     * @param who 调用者
     * @param taskParam 任务参数
     */
    @Override
    public void sendMsg2Kafka(int taskType, String who,TaskParam taskParam){
        sendMsg2Kafka(taskType,who,taskParam,1);
    }

    /**
     * 查询当前有最大BloggerMain的id值，用于代表记录数
     * @return
     */
    @Override
    public long objectTotals() {
        return 1;
    }

    /**
     *
     * @param messages
     * @param topic
     */
    @Override
    public void sendMsg(List<InsBloggerMessage> messages, String topic) {
        for(int i = 0;i < taskConfig.getInsAnalysisAccountCheckUnusualSendMsgCount(); ++i){
            messages.forEach( msg -> insBloggerMsgBaseInfoProducer.send(taskConfig.getInsAnalysisAccountCheckUnusualOnlineTopic(),msg));
        }
        XxlJobLogger.log("投递主题 = {},投递消息数 = {}", taskConfig.getInsAnalysisAccountCheckUnusualOnlineTopic(),messages.size());
        log.info("投递主题 = {},投递消息数 = {}", taskConfig.getInsAnalysisAccountCheckUnusualOnlineTopic(),messages.size());

    }
}
